package com.lagou.work6;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import static org.apache.flink.table.api.Expressions.$;

/*
    flink table案例
 */
public class TableDemo {
    public static void main(String[] args) throws Exception {
        //获取flink执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //创建tenv
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        //获取流式数据
        DataStreamSource<Tuple2<String,Integer>> data = env.addSource(
                new SourceFunction<Tuple2<String, Integer>>() {
                    int num = 0;
                    @Override
                    public void run(SourceContext<Tuple2<String, Integer>> sourceContext) throws Exception {
                        while (true){
                            sourceContext.collect(new Tuple2<>("target-" + num, num));
                            num++;
                            Thread.sleep(1000);
                        }
                    }

                    @Override
                    public void cancel() {

                    }
                }
        );
        //将流式数据源作为table
        Table table = tEnv.fromDataStream(data, $("target"), $("num"));
        //对table数据做查询
        Table selectTable = table.select($("target"));
        //将table转为数据流
        DataStream<Tuple2<Boolean, Row>> res = tEnv.toRetractStream(selectTable, Row.class);
        //打印输出
        res.print();
        env.execute();
    }
}
